package org.cybzacg.encryption.batch.impl;

import org.cybzacg.encryption.batch.BatchProcessor;
import org.cybzacg.encryption.context.EncryptionContext;
import org.cybzacg.encryption.model.EncryptionResult;
import org.cybzacg.encryption.model.ValidationResult;
import org.cybzacg.encryption.utils.ValidationUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 并行批处理器实现
 * 使用多线程并行处理批量数据
 * 
 * @author encryption team
 * @version 1.0
 */
@Slf4j
public class ParallelBatchProcessor<T, R> implements BatchProcessor<T, R> {

    // 配置参数
    private volatile int batchSize = 100;
    private volatile int maxConcurrency = Runtime.getRuntime().availableProcessors();
    private volatile long timeout = 30000; // 30秒

    // 线程池
    private final ExecutorService executorService;

    // 统计信息
    private final AtomicLong totalBatches = new AtomicLong(0);
    private final AtomicLong totalProcessed = new AtomicLong(0);
    private final AtomicLong totalSuccess = new AtomicLong(0);
    private final AtomicLong totalFailures = new AtomicLong(0);
    private final AtomicLong totalProcessingTime = new AtomicLong(0);
    private volatile long maxProcessingTime = 0;
    private volatile long minProcessingTime = Long.MAX_VALUE;

    // 监听器
    private final List<BatchListener<T, R>> listeners = new CopyOnWriteArrayList<>();

    /**
     * 构造函数
     */
    public ParallelBatchProcessor() {
        this.executorService = Executors.newFixedThreadPool(maxConcurrency);
    }

    /**
     * 构造函数
     * 
     * @param maxConcurrency 最大并发数
     */
    public ParallelBatchProcessor(int maxConcurrency) {
        this.maxConcurrency = maxConcurrency;
        this.executorService = Executors.newFixedThreadPool(maxConcurrency);
    }

    /**
     * 构造函数
     * 
     * @param executorService 自定义线程池
     */
    public ParallelBatchProcessor(ExecutorService executorService) {
        this.executorService = executorService;
    }

    @Override
    public List<R> processBatch(List<T> dataList, EncryptionContext context, Function<T, R> processor) {
        if (dataList == null || dataList.isEmpty()) {
            return Collections.emptyList();
        }

        long startTime = System.currentTimeMillis();

        // 验证配置
        ValidationResult validationResult = validateConfiguration(context);
        if (!validationResult.isValid()) {
            throw new IllegalArgumentException("Invalid configuration: " + validationResult.getMessage());
        }

        // 通知开始
        notifyBatchStart(dataList.size());

        try {
            // 分批处理
            List<R> results = new ArrayList<>(dataList.size());
            List<CompletableFuture<R>> futures = new ArrayList<>();

            for (int i = 0; i < dataList.size(); i += batchSize) {
                int endIndex = Math.min(i + batchSize, dataList.size());
                List<T> batch = dataList.subList(i, endIndex);

                CompletableFuture<R> future = CompletableFuture.supplyAsync(() -> {
                    return processSingleBatch(batch, context, processor);
                }, executorService);

                futures.add(future);
            }

            // 等待所有任务完成
            for (CompletableFuture<R> future : futures) {
                try {
                    R result = future.get(timeout, TimeUnit.MILLISECONDS);
                    if (result != null) {
                        results.add(result);
                    }
                } catch (Exception e) {
                    // 记录失败
                    totalFailures.incrementAndGet();
                }
            }

            // 更新统计信息
            long processingTime = System.currentTimeMillis() - startTime;
            updateStatistics(processingTime, dataList.size(), results.size());

            // 通知完成
            notifyBatchComplete(createBatchResult(results, dataList.size(), processingTime));

            return results;

        } catch (Exception e) {
            // 通知错误
            notifyBatchError(createBatchError(e));
            throw new RuntimeException("Batch processing failed", e);
        }
    }

    @Override
    public CompletableFuture<List<R>> processBatchAsync(List<T> dataList, EncryptionContext context,
            Function<T, R> processor) {
        return CompletableFuture.supplyAsync(() -> processBatch(dataList, context, processor));
    }

    @Override
    public List<EncryptionResult> encryptBatch(List<byte[]> dataList, EncryptionContext context) {
        List<EncryptionResult> results = new ArrayList<>(dataList.size());
        List<CompletableFuture<EncryptionResult>> futures = new ArrayList<>();

        for (int i = 0; i < dataList.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, dataList.size());
            List<byte[]> batch = dataList.subList(i, endIndex);

            CompletableFuture<EncryptionResult> future = CompletableFuture.supplyAsync(() -> {
                return processSingleEncryptionBatch(batch, context);
            }, executorService);

            futures.add(future);
        }

        // 等待所有任务完成
        for (CompletableFuture<EncryptionResult> future : futures) {
            try {
                EncryptionResult result = future.get(timeout, TimeUnit.MILLISECONDS);
                if (result != null) {
                    results.add(result);
                }
            } catch (Exception e) {
                // 记录失败
                totalFailures.incrementAndGet();
            }
        }

        return results;
    }

    @Override
    public CompletableFuture<List<EncryptionResult>> encryptBatchAsync(List<byte[]> dataList,
            EncryptionContext context) {
        return CompletableFuture.supplyAsync(() -> encryptBatch(dataList, context));
    }

    @Override
    public List<byte[]> decryptBatch(List<byte[]> dataList, EncryptionContext context) {
        List<byte[]> results = new ArrayList<>(dataList.size());
        List<CompletableFuture<byte[]>> futures = new ArrayList<>();

        for (int i = 0; i < dataList.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, dataList.size());
            List<byte[]> batch = dataList.subList(i, endIndex);

            CompletableFuture<byte[]> future = CompletableFuture.supplyAsync(() -> {
                return processSingleDecryptionBatch(batch, context);
            }, executorService);

            futures.add(future);
        }

        // 等待所有任务完成
        for (CompletableFuture<byte[]> future : futures) {
            try {
                byte[] result = future.get(timeout, TimeUnit.MILLISECONDS);
                if (result != null) {
                    results.add(result);
                }
            } catch (Exception e) {
                // 记录失败
                totalFailures.incrementAndGet();
            }
        }

        return results;
    }

    @Override
    public CompletableFuture<List<byte[]>> decryptBatchAsync(List<byte[]> dataList, EncryptionContext context) {
        return CompletableFuture.supplyAsync(() -> decryptBatch(dataList, context));
    }

    @Override
    public int getBatchSize() {
        return batchSize;
    }

    @Override
    public void setBatchSize(int batchSize) {
        if (batchSize <= 0) {
            throw new IllegalArgumentException("Batch size must be positive");
        }
        this.batchSize = batchSize;
    }

    @Override
    public int getMaxConcurrency() {
        return maxConcurrency;
    }

    @Override
    public void setMaxConcurrency(int maxConcurrency) {
        if (maxConcurrency <= 0) {
            throw new IllegalArgumentException("Max concurrency must be positive");
        }
        this.maxConcurrency = maxConcurrency;
    }

    @Override
    public long getTimeout() {
        return timeout;
    }

    @Override
    public void setTimeout(long timeout) {
        if (timeout <= 0) {
            throw new IllegalArgumentException("Timeout must be positive");
        }
        this.timeout = timeout;
    }

    @Override
    public ValidationResult validateConfiguration(EncryptionContext context) {
        if (context == null) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Context cannot be null")
                    .build();
        }

        if (batchSize <= 0) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Batch size must be positive")
                    .build();
        }

        if (maxConcurrency <= 0) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Max concurrency must be positive")
                    .build();
        }

        if (timeout <= 0) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Timeout must be positive")
                    .build();
        }

        return new ValidationResult.Builder()
                .valid(true)
                .message("Configuration is valid")
                .build();
    }

    @Override
    public BatchStatistics getStatistics() {
        return new BatchStatisticsImpl(
                totalBatches.get(),
                totalProcessed.get(),
                totalSuccess.get(),
                totalFailures.get(),
                totalProcessed.get() > 0 ? (double) totalProcessingTime.get() / totalProcessed.get() : 0.0,
                maxProcessingTime,
                minProcessingTime == Long.MAX_VALUE ? 0 : minProcessingTime,
                calculateThroughput(),
                totalProcessed.get() > 0 ? (double) totalSuccess.get() / totalProcessed.get() : 0.0);
    }

    @Override
    public void resetStatistics() {
        totalBatches.set(0);
        totalProcessed.set(0);
        totalSuccess.set(0);
        totalFailures.set(0);
        totalProcessingTime.set(0);
        maxProcessingTime = 0;
        minProcessingTime = Long.MAX_VALUE;
    }

    /**
     * 添加批处理监听器
     * 
     * @param listener 监听器
     */
    public void addListener(BatchListener<T, R> listener) {
        if (listener != null) {
            listeners.add(listener);
        }
    }

    /**
     * 移除批处理监听器
     * 
     * @param listener 监听器
     */
    public void removeListener(BatchListener<T, R> listener) {
        listeners.remove(listener);
    }

    /**
     * 关闭批处理器
     */
    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 处理单个批次
     * 
     * @param batch     批次数据
     * @param context   加密上下文
     * @param processor 处理函数
     * @return 处理结果（仅返回第一个结果，用于兼容当前接口）
     */
    private R processSingleBatch(List<T> batch, EncryptionContext context, Function<T, R> processor) {
        if (batch.isEmpty()) {
            return null;
        }

        // 处理批次中的所有元素
        List<R> results = batch.stream()
                .map(processor)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());

        // 为了兼容当前方法签名，返回第一个结果
        // TODO: 考虑将返回类型改为 List<R>
        return results.isEmpty() ? null : results.get(0);
    }

    /**
     * 处理单个加密批次
     * 
     * <p>
     * 注意：此方法需要注入 EncryptionEngine 才能完整实现。
     * 当前实现为占位符，不执行实际加密操作。
     * 建议使用 processBatch 方法并传入加密处理函数。
     * 
     * @param batch   待加密数据批次
     * @param context 加密上下文
     * @return 加密结果（占位符实现）
     * @deprecated 使用 processBatch 方法代替
     */
    @Deprecated
    private EncryptionResult processSingleEncryptionBatch(List<byte[]> batch, EncryptionContext context) {
        if (batch.isEmpty()) {
            return null;
        }

        log.warn("processSingleEncryptionBatch 方法未完整实现，需要注入 EncryptionEngine。" +
                "建议使用 processBatch 方法并传入加密处理函数。");

        try {
            // TODO: 需要注入 EncryptionEngine 来实现真正的加密
            // 当前为占位符实现，仅返回第一个元素
            byte[] data = batch.get(0);
            return new EncryptionResult.Builder()
                    .data(data)
                    .status(org.cybzacg.encryption.enums.EncryptionStatus.SUCCESS)
                    .algorithm(context.getAlgorithm())
                    .build();
        } catch (Exception e) {
            log.error("批量加密处理失败", e);
            return new EncryptionResult.Builder()
                    .status(org.cybzacg.encryption.enums.EncryptionStatus.UNKNOWN_ERROR)
                    .errorMessage(e.getMessage())
                    .algorithm(context.getAlgorithm())
                    .build();
        }
    }

    /**
     * 处理单个解密批次
     * 
     * <p>
     * 注意：此方法需要注入 EncryptionEngine 才能完整实现。
     * 当前实现为占位符，不执行实际解密操作。
     * 建议使用 processBatch 方法并传入解密处理函数。
     * 
     * @param batch   待解密数据批次
     * @param context 解密上下文
     * @return 解密结果（占位符实现）
     * @deprecated 使用 processBatch 方法代替
     */
    @Deprecated
    private byte[] processSingleDecryptionBatch(List<byte[]> batch, EncryptionContext context) {
        if (batch.isEmpty()) {
            return null;
        }

        log.warn("processSingleDecryptionBatch 方法未完整实现，需要注入 EncryptionEngine。" +
                "建议使用 processBatch 方法并传入解密处理函数。");

        try {
            // TODO: 需要注入 EncryptionEngine 来实现真正的解密
            // 当前为占位符实现，仅返回第一个元素
            return batch.get(0);
        } catch (Exception e) {
            log.error("批量解密处理失败", e);
            return null;
        }
    }

    /**
     * 更新统计信息
     */
    private void updateStatistics(long processingTime, int totalCount, int successCount) {
        totalBatches.incrementAndGet();
        totalProcessed.addAndGet(totalCount);
        totalSuccess.addAndGet(successCount);
        totalFailures.addAndGet(totalCount - successCount);
        totalProcessingTime.addAndGet(processingTime);

        if (processingTime > maxProcessingTime) {
            maxProcessingTime = processingTime;
        }

        if (processingTime < minProcessingTime) {
            minProcessingTime = processingTime;
        }
    }

    /**
     * 计算吞吐量
     */
    private double calculateThroughput() {
        long totalTime = totalProcessingTime.get();
        if (totalTime > 0) {
            return (double) totalProcessed.get() * 1000 / totalTime; // 操作/秒
        }
        return 0.0;
    }

    /**
     * 创建批处理结果
     */
    private BatchResult<R> createBatchResult(List<R> results, int totalCount, long processingTime) {
        return new BatchResultImpl<>(results, totalCount, processingTime);
    }

    /**
     * 创建批处理错误
     */
    private BatchError<R> createBatchError(Exception exception) {
        return new BatchErrorImpl<>(exception);
    }

    /**
     * 通知批处理开始
     */
    private void notifyBatchStart(int totalCount) {
        for (BatchListener<T, R> listener : listeners) {
            try {
                listener.onBatchStart(totalCount);
            } catch (Exception e) {
                // 忽略监听器异常
            }
        }
    }

    /**
     * 通知批处理完成
     */
    private void notifyBatchComplete(BatchResult<R> result) {
        for (BatchListener<T, R> listener : listeners) {
            try {
                listener.onBatchComplete(result);
            } catch (Exception e) {
                // 忽略监听器异常
            }
        }
    }

    /**
     * 通知批处理错误
     */
    private void notifyBatchError(BatchError<R> error) {
        for (BatchListener<T, R> listener : listeners) {
            try {
                listener.onBatchError(error);
            } catch (Exception e) {
                // 忽略监听器异常
            }
        }
    }

    /**
     * 批处理结果实现
     */
    private static class BatchResultImpl<R> implements BatchResult<R> {

        private final List<R> successResults;
        private final int totalCount;
        private final long processingTime;

        public BatchResultImpl(List<R> successResults, int totalCount, long processingTime) {
            this.successResults = successResults != null ? successResults : Collections.emptyList();
            this.totalCount = totalCount;
            this.processingTime = processingTime;
        }

        @Override
        public List<R> getSuccessResults() {
            return successResults;
        }

        @Override
        public List<BatchError<R>> getFailedResults() {
            return Collections.emptyList(); // 简化实现
        }

        @Override
        public int getTotalCount() {
            return totalCount;
        }

        @Override
        public int getSuccessCount() {
            return successResults.size();
        }

        @Override
        public int getFailureCount() {
            return totalCount - successResults.size();
        }

        @Override
        public double getSuccessRate() {
            return totalCount > 0 ? (double) getSuccessCount() / totalCount : 0.0;
        }

        @Override
        public long getProcessingTime() {
            return processingTime;
        }

        @Override
        public boolean isAllSuccess() {
            return getFailureCount() == 0;
        }

        @Override
        public boolean hasFailures() {
            return getFailureCount() > 0;
        }
    }

    /**
     * 批处理错误实现
     */
    private static class BatchErrorImpl<R> implements BatchError<R> {

        private final Exception exception;

        public BatchErrorImpl(Exception exception) {
            this.exception = exception;
        }

        @Override
        public Object getOriginalData() {
            return null;
        }

        @Override
        public String getErrorMessage() {
            return exception != null ? exception.getMessage() : "Unknown error";
        }

        @Override
        public Exception getException() {
            return exception;
        }

        @Override
        public ErrorType getErrorType() {
            return ErrorType.PROCESSING_ERROR;
        }
    }

    /**
     * 批处理统计信息实现
     */
    private static class BatchStatisticsImpl implements BatchStatistics {

        private final long totalBatches;
        private final long totalProcessed;
        private final long totalSuccess;
        private final long totalFailures;
        private final double averageProcessingTime;
        private final long maxProcessingTime;
        private final long minProcessingTime;
        private final double throughput;
        private final double averageSuccessRate;

        public BatchStatisticsImpl(long totalBatches, long totalProcessed, long totalSuccess,
                long totalFailures, double averageProcessingTime,
                long maxProcessingTime, long minProcessingTime,
                double throughput, double averageSuccessRate) {
            this.totalBatches = totalBatches;
            this.totalProcessed = totalProcessed;
            this.totalSuccess = totalSuccess;
            this.totalFailures = totalFailures;
            this.averageProcessingTime = averageProcessingTime;
            this.maxProcessingTime = maxProcessingTime;
            this.minProcessingTime = minProcessingTime;
            this.throughput = throughput;
            this.averageSuccessRate = averageSuccessRate;
        }

        @Override
        public long getTotalBatches() {
            return totalBatches;
        }

        @Override
        public long getTotalProcessed() {
            return totalProcessed;
        }

        @Override
        public long getTotalSuccess() {
            return totalSuccess;
        }

        @Override
        public long getTotalFailures() {
            return totalFailures;
        }

        @Override
        public double getAverageProcessingTime() {
            return averageProcessingTime;
        }

        @Override
        public long getMaxProcessingTime() {
            return maxProcessingTime;
        }

        @Override
        public long getMinProcessingTime() {
            return minProcessingTime;
        }

        @Override
        public double getThroughput() {
            return throughput;
        }

        @Override
        public double getAverageSuccessRate() {
            return averageSuccessRate;
        }

        @Override
        public String toString() {
            return String.format(
                    "BatchStatistics:\n" +
                            "  Total Batches: %d\n" +
                            "  Total Processed: %d\n" +
                            "  Success: %d, Failures: %d\n" +
                            "  Success Rate: %.2f%%\n" +
                            "  Avg Processing Time: %.2fms\n" +
                            "  Max/Min Time: %dms/%dms\n" +
                            "  Throughput: %.2f ops/sec",
                    totalBatches, totalProcessed, totalSuccess, totalFailures,
                    averageSuccessRate * 100, averageProcessingTime,
                    maxProcessingTime, minProcessingTime, throughput);
        }
    }
}
